#include "threads.h"
#include "clog.h"

using namespace std;

threads::threads(){
    threads(8);
}

threads::threads(int _max){
    max_thhreadNum =  _max;
    numRunningOneShot = 0;
    jobCount = 0;
    unsigned int err = 0;

    err |= sem_init(&queueMutex,0,1);
    err |= sem_init(&semTasks,0,0);
    err |= sem_init(&alljobs,0,0);
    numRunningJob = 0;
    numThread = 0;

    stopThread = false;
}

/***
 *      Static Function. Scheular
 */
void* threads::threadFunc(void *p){
    threads     * th = ((threadP*)p)->th;
    worker      * w;
    clog::d("New Thread ");
    clog::dl( ((threadP*)p)->threadNum );
    while(1){
        sem_wait(&th->semTasks);
        if(th->stopThread){
            sem_post(&th->semTasks);
            break;
        }
        sem_wait(&th->queueMutex);
            w  = th->tasks.top();
            th->tasks.pop();
            th->runningTasks.insert(w);
        sem_post(&th->queueMutex);
        w->run();
        sem_wait(&th->queueMutex);
            th->numRunningJob--;
            th->runningTasks.erase(w);
        if (w->workerType == WORKER_TYPE_ONESHOT){
            sem_post(&th->alljobs);
            th->numRunningOneShot--;
        }
        sem_post(&th->queueMutex);
    }
    clog::d("Thread ");
    clog::d( ((threadP*)p)->threadNum );
    clog::dl(" EXIT");
    return NULL;
}

void threads::wait(){
    clog::d("Waiting:");
    clog::dl(jobCount);
    for(;jobCount > 0; jobCount--){
        sem_wait(&alljobs);
    }
    clog::dl("Waiting Finished.");
}

bool threads::createThread(void){
    threadP* threadParam = new threadP;
    threadParam->th = this;
    threadParam->threadNum = numThread;

    unsigned int err;
    pthread_t tid;
    err = pthread_create(&tid,NULL,threadFunc,threadParam);
    if(err){
        clog::el("ERROR. ERR when create thread");
        return false;
    }
    tids.push_back(tid);
    clog::dl("threads created");
    numThread++;
    return true;
}

void threads::stop(void){
    sem_wait(&queueMutex);
        for(set<worker*>::iterator i = runningTasks.begin();i != runningTasks.end();i++){
            (*i)->stop();
        }
        clog::dl("finished call stop function");
    sem_post(&queueMutex);
    wait();
    stopThread = true;
    sem_post(&semTasks);
    clog::dl("calling join function");
    for(unsigned int i = 0; i < tids.size() ; i++){
        pthread_join(tids[i],NULL);
    }
    clog::dl("finished stop");
}

// TODO change code here to garentee that perm job are executed immerdiately
// while one short jobs can be called later
void threads::addOneshot(worker *w){
    sem_wait(&queueMutex);
        clog::d("Add Oneshot Job:");
        clog::dl(numRunningOneShot);
        w->workerType = WORKER_TYPE_ONESHOT;
        tasks.push(w);
        jobCount++;
        numRunningOneShot++;
        if(numRunningOneShot > max_thhreadNum){
            sem_post(&queueMutex);
            sem_post(&semTasks);
            return;
        }
        createThread();
    sem_post(&queueMutex);
    sem_post(&semTasks);
}

/**
  * ADD PERM JOBS that may run for a very long time, So each worker needs
  * a extra thread. I assume that this kind of job will never exit or will
  * only exit after a surfercent long time. And the number of this kind of
  * job are limited.
  */
// TODO --- add code here.
void threads::addJob(worker *w){
    sem_wait(&queueMutex);
        clog::d("Add Perm Job:");
        clog::dl(numRunningJob);
        tasks.push(w);
        numRunningJob++;
        if(numThread > numRunningJob+max_thhreadNum){
            sem_post(&semTasks);
            sem_post(&queueMutex);
            return;
        }
        createThread();
        sem_post(&semTasks);
    sem_post(&queueMutex);
}
